package com.jacquesh;

import org.apache.flink.api.common.functions.FilterFunction;
import org.apache.flink.api.common.functions.RichFilterFunction;
import org.apache.flink.api.common.functions.RichFlatMapFunction;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.util.Collector;

public class FlinkDemo0011 {
    public static void main(String[] args) throws Exception {

        StreamExecutionEnvironment environment =
                StreamExecutionEnvironment.getExecutionEnvironment();
        DataStreamSource<String> stream = environment
                .socketTextStream("192.168.88.180", 8888);
         stream.flatMap(new RichFlatMapFunction<String, Long>() {
             @Override
             public void flatMap(String s, Collector<Long> collector) throws Exception {
                 collector.collect(Long.parseLong(s));
             }
         }).filter(new RichFilterFunction<Long>() {
             @Override
             public void open(Configuration parameters)  {
                 /**
                  * 由AbstractRichFunction提供的getRuntimeContext()方法
                  * 可以获取运行时候的上下文信息
                  * **/
                 System.out.println(getRuntimeContext().getTaskName());
                 System.out.println(getRuntimeContext().getTaskNameWithSubtasks());
             }
            @Override
            public boolean filter(Long s)  {
                return s<100;
            }
         }).print();

        environment.execute("1111");
    }
}
